package com.disruptor.sample._07advance.run;

import com.disruptor.sample._07advance.consumer.TradeHandler;
import com.disruptor.sample._07advance.dto.Trade;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkerPool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class AdvanceMain02 {

  public static RingBuffer<Trade> buildRingBuffer() {
    int BUFFER_SIZE = 1024;
    int THREAD_NUMBERS = 4;

    EventFactory<Trade> eventFactory = new EventFactory<Trade>() {
      public Trade newInstance() {
        return new Trade();
      }
    };

    RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);
    return ringBuffer;
  }

  public static ExecutorService buildProcessExecutors() {
    // 创建线程池
    int THREAD_NUMBERS = 4;
    ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);
    return executors;
  }

  public static WorkerPool<Trade> buildProcessors(ExecutorService executors,
      final RingBuffer<Trade> ringBuffer, SequenceBarrier sequenceBarrier) {
    // 注意改写法是 同一个消息时，WorkerPool 中有多个消费者，只有一个消费者进行处理
    WorkerPool<Trade> workerPool = new WorkerPool<Trade>(ringBuffer, sequenceBarrier,
        new IgnoreExceptionHandler(), new TradeHandler(), new TradeHandler(), new TradeHandler());

    workerPool.start(executors);

    return workerPool;
  }

  public static void main(String[] args) throws InterruptedException {
    final RingBuffer<Trade> ringBuffer = buildRingBuffer();

    ExecutorService executors = buildProcessExecutors();

    // 创建 SequenceBarrier, 用于平衡生产者和消费者速率，用障碍来处理
    SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

    WorkerPool<Trade> workerPool = buildProcessors(executors, ringBuffer,
        sequenceBarrier);

    //下面这个生产8个数据
    for (int i = 0; i < 8; i++) {
      long seq = ringBuffer.next();
      ringBuffer.get(seq).setPrice(Math.random() * 9999);
      ringBuffer.publish(seq);
    }
    Thread.sleep(5000);
    // 通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）
    workerPool.halt();
    executors.shutdown();
  }

}
